package cn.monkey.transport.mq.redisson;

import cn.money.transport.mq.MqInboundConfig;
import org.redisson.api.*;
import org.redisson.client.codec.Codec;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Predicate;

public class SimpleRedissonInbound implements RedissonInbound {
    private final MqInboundConfig inboundConfig;
    private final AtomicBoolean closed;
    private final Flux<Object> receiver;

    public SimpleRedissonInbound(RedissonClient redissonClient,
                                 Codec codec,
                                 MqInboundConfig inboundConfig) {
        this.inboundConfig = inboundConfig;
        this.closed = new AtomicBoolean(false);
        this.receiver = Flux.fromArray(inboundConfig.getTopics())
                .map(topic -> redissonClient.getBlockingQueue(topic, codec))
                .flatMap(blockQueue -> Flux.create(sink -> {
                    Object lock = new Object();
                    while (!this.closed.get()) {
                        CompletableFuture<List<Object>> completableFuture = blockQueue.pollAsync(this.inboundConfig.getPollSize())
                                .toCompletableFuture();
                        completableFuture.thenAccept(objects -> {
                            if (objects == null || objects.isEmpty()) {
                                synchronized (lock) {
                                    try {
                                        lock.wait(this.inboundConfig.getPollFrequencyMs());
                                    } catch (InterruptedException e) {
                                        sink.error(e);
                                    }
                                }
                                return;
                            }
                            for (Object o : objects) {
                                sink.next(o);
                            }
                            synchronized (lock) {
                                lock.notifyAll();
                            }
                        });
                        synchronized (lock) {
                            try {
                                lock.wait(this.inboundConfig.getPollFrequencyMs());
                            } catch (InterruptedException e) {
                                sink.error(e);
                                break;
                            }
                        }
                    }
                    sink.complete();
                }));
    }

    @Override
    public <T> Flux<T> receive(Class<T> clazz, Predicate<T> p) {
        return this.receiver.filter(clazz::isInstance)
                .map(clazz::cast)
                .filter(p);
    }

    @Override
    public Mono<Void> close() {
        this.closed.set(true);
        return Mono.empty();
    }

    @Override
    public MqInboundConfig inboundConfig() {
        return this.inboundConfig;
    }
}
